package mq

import (
	"bytes"
	"fmt"

	"github.com/streadway/amqp"
)

// Callback 定义回调函数类型
type Callback func(msg string)

// Connect 连接mq服务器
func Connect() (*amqp.Connection, error) {
	conn, err := amqp.Dial("amqp://admin:admin@192.168.176.133:5672/")
	return conn, err
}

// Publish 发布者
func Publish(exchange string, queueName string, body string) error {
	// 建立连接
	conn, err := Connect()
	if err != nil {
		return err
	}
	defer conn.Close()

	// 创建通道
	channel, err := conn.Channel()
	if err != nil {
		return err
	}
	defer channel.Close()

	// 创建队列
	q, err := channel.QueueDeclare(
		queueName,
		true, // 是否持久化
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		return err
	}

	// 发送消息
	err = channel.Publish(exchange, q.Name, false, false, amqp.Publishing{
		DeliveryMode: amqp.Persistent,
		ContentType:  "text/plain",
		Body:         []byte(body),
	})
	return err
}

// Consumer 收受者方法、
func Consumer(exchange string, queueName string, callback Callback) {
	// 建立连接
	conn, err := Connect()
	defer conn.Close()
	if err != nil {
		fmt.Println(err)
		return
	}

	// 创建通告channel
	channel, err := conn.Channel()
	defer channel.Close()
	if err != nil {
		fmt.Println(err)
		return
	}
	// 创建queue
	q, err := channel.QueueDeclare(
		queueName,
		true,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		fmt.Println(err)
		return
	}

	msgs, err := channel.Consume(q.Name, "", true, false, false, false, nil)
	if err != nil {
		fmt.Println(err)
		return
	}
	// 定义一个管道
	forever := make(chan bool)
	go func() {
		for d := range msgs {
			s := BytesToString(&(d.Body))
			callback(*s)
		}
	}()
	fmt.Println("Waiting for messages")
	<-forever
}

// BytesToString 类型转换
func BytesToString(b *[]byte) *string {
	s := bytes.NewBuffer(*b)
	r := s.String()
	return &r
}
